package com.zzq.demo.kafka.mq.listener;

import com.zzq.demo.kafka.dto.DataDownDTO;
import com.zzq.demo.kafka.dto.DataUpDTO;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @ description:
 * @ author: zhaoziqian
 * @ date: 2019/1/7
 */
@Slf4j
@Component
public class DownListener {

    @KafkaListener(topics = "kds_test_up")
    public void listen(ConsumerRecord<String, DataUpDTO> record, Acknowledgment ack) {
        log.info("【MESSAGE DOWN】接收到数据下发信息，分区 ： {} ， 消息key : {} ", record.partition(), record.key());
        try {
            System.out.println(record.value());
        } catch (Exception e) {
            log.error("数据下发错误，信息 key ：" + record.key(), e);
        } finally {
            ack.acknowledge();
        }
    }
}
